package com.lzq.mq.broker;



import com.lzq.mq.broker.cache.CommonCache;
import com.lzq.mq.broker.config.ConsumeQueueOffsetLoader;
import com.lzq.mq.broker.config.GlobalPropertiesLoader;
import com.lzq.mq.broker.config.MqTopicLoader;
import com.lzq.mq.broker.core.CommitLogAppendHandler;
import com.lzq.mq.broker.core.ConsumeQueueAppendHandler;
import com.lzq.mq.broker.model.MqTopicModel;
import com.lzq.mq.broker.netty.broker.BrokerServer;
import com.lzq.mq.broker.netty.nameserver.NameServerClient;
import com.lzq.mq.broker.slave.SlaveSyncService;
import com.lzq.mq.common.enums.BrokerClusterModeEnum;
import com.lzq.mq.common.enums.BrokerRegistryRoleEnum;
import com.lzq.mq.common.event.EventBus;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.List;
import java.util.Objects;

@Slf4j
public class BrokerStartUp {

    private static MqTopicLoader topicLoader;
    private static GlobalPropertiesLoader globalPropertiesLoader;
    private static ConsumeQueueOffsetLoader consumeQueueOffsetLoader;
    private static CommitLogAppendHandler messageAppendHandler;
    private static ConsumeQueueAppendHandler consumeQueueAppendHandler;

    private static BrokerServer brokerServer;

    public static void main(String[] args) throws IOException, InterruptedException {

        // 加载配置和数据
        loadProperties();

        // 连接nameserver, 注册服务
        initNameServerChannel();

        // 初始化消息总线
        EventBus.init();

        // 初始化Broker服务, 开始监听请求
        initBrokerServer();     // 注意必须最后调用, 因为其会阻塞主线程
    }

    /**
     * 加载配置, 导入主题信息, 导入主题
     */
    private static void loadProperties() {
        globalPropertiesLoader = new GlobalPropertiesLoader();
        topicLoader = new MqTopicLoader();
        consumeQueueOffsetLoader = new ConsumeQueueOffsetLoader();
        messageAppendHandler = new CommitLogAppendHandler();
        consumeQueueAppendHandler = new ConsumeQueueAppendHandler();

        globalPropertiesLoader.loadProperties();    // 加载全局配置
        topicLoader.loadProperties();               // 加载主题
        topicLoader.startRefreshMqTopicInfoTask();  // 开启刷新主题信息任务
        consumeQueueOffsetLoader.loadProperties();  // 加载消费队列偏移量
        consumeQueueOffsetLoader.startRefreshConsumeQueueOffsetTask(); // 开启刷新消费队列偏移量任务

        // 加载主题映射
        List<MqTopicModel> topicModels = CommonCache.getMqTopicModelList();
        for (MqTopicModel topicModel : topicModels) {
            String topicName = topicModel.getTopicName();
            messageAppendHandler.prepareMMapLoading(topicName);
            consumeQueueAppendHandler.prepareConsumeQueue(topicName);
        }

        CommonCache.setConsumeQueueAppendHandler(consumeQueueAppendHandler);
    }

    /**
     * 初始化NameServer通道, 注册服务
     */
    private static void initNameServerChannel() {
        NameServerClient nameServerClient = new NameServerClient();
        CommonCache.setNameServerClient(nameServerClient);
        nameServerClient.initConnect();
        nameServerClient.sendRegistryMsg();     // 向nameserver注册服务
        // 如果是主节点直接结束即可
        if (!Objects.equals(CommonCache.getGlobalProperties().getBrokerClusterMode(), BrokerClusterModeEnum.MASTER_SLAVE.getCode())
                || !Objects.equals(CommonCache.getGlobalProperties().getBrokerClusterRole(), BrokerRegistryRoleEnum.SLAVE.getCode())) {
            return;
        }
        // 获取主节点, 连接主节点
        String masterAddress = nameServerClient.queryBrokerMasterAddress();
        if(masterAddress == null) {
            log.error("未获取到主节点地址");
        }
        //尝试链接主broker
        SlaveSyncService slaveSyncService = new SlaveSyncService();
        CommonCache.setSlaveSyncService(slaveSyncService);
        boolean connectionStat = false;
        if (masterAddress != null) {
            connectionStat = slaveSyncService.connectMasterBrokerNode(masterAddress);
        }
        if(connectionStat) {
            slaveSyncService.sendStartSyncMsg();
        }
    }

    /**
     * 初始化Broker服务
     * 此方法必须最后调用, 其会阻塞后续命令
     */
    public static void initBrokerServer() throws InterruptedException {

        // 启动重平衡任务
        CommonCache.getReBalanceInstancePool().startReBalanceJob();

        Integer brokerPort = CommonCache.getGlobalProperties().getBrokerPort();
        BrokerServer brokerServer = new BrokerServer();
        brokerServer.startServer(brokerPort);
        // 添加钩子函数, 关闭程序时停止brokerServer
        // Runtime.getRuntime().addShutdownHook(new Thread(brokerServer::stopServer));
    }
}
